package com.stackfarm.fvservice.mqtt;

import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import com.stackfarm.fvcommon.utils.MqttTools;
import com.stackfarm.fvcommon.utils.SpringUtils;
import com.stackfarm.fvservice.config.InfluxDBConfig;
import org.apache.commons.codec.binary.Hex;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.Resource;
import java.text.SimpleDateFormat;
import java.util.Base64;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * MQTT消息处理类
 *
 * @author zhongyulin
 */
public class TopMsgCallback implements MqttCallback {

    private static Logger logger = LoggerFactory.getLogger(TopMsgCallback.class);

    private MqttClient client;
    private MqttConnectOptions options;
    private String[] topic;
    private int[] qos;

    public TopMsgCallback() {
    }

    public TopMsgCallback(MqttClient client, MqttConnectOptions options, String[] topic, int[] qos) {
        this.client = client;
        this.options = options;
        this.topic = topic;
        this.qos = qos;
    }

    /**
     * 断开重连
     */
    public void connectionLost(Throwable cause) {
        logger.info("MQTT连接断开，发起重连" + cause);
        while (true) {
            try {
                Thread.sleep(30000);
                client.connect(options);
                //订阅消息
                client.subscribe(topic, qos);
                logger.info("MQTT重新连接成功:" + client);
                break;
            } catch (Exception e) {
                e.printStackTrace();
                continue;
            }
        }
    }

    /**
     * 接收到消息调用令牌中调用
     */
    public void deliveryComplete(IMqttDeliveryToken token) {
    }

    /**
     * 消息处理
     */
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        InfluxDBConfig influxDBConfig = SpringUtils.getBean("influxdbConfig");
        System.out.println();
        //订阅消息字符
        String msg = new String(message.getPayload());
        logger.info("topic:" + topic);
        logger.info("msg:" + msg);
        if (msg.compareTo("close") != 0) {
            JsonParser jp = new JsonParser();
            //将json字符串转化成json对象
            JsonObject jo = jp.parse(msg).getAsJsonObject();
            //获取message对应的值
            byte[] data = Base64.getDecoder().decode(jo.get("data").getAsString());
            String datas = new String(Hex.encodeHex(data));
            String termid = jo.get("devEUI").getAsString();
            logger.info("data" + datas);
            logger.info("eui" + termid);
            //输出时间用于查看用
            SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//设置日期格式
            System.out.println(df.format(new Date()));
            //解析data
            if ("8d00003100000001".equals(termid) && !datas.substring(8, 10).equals("00")) {//如果是小于
                //光照度 illumination
                int pos = 10;
                int len = Integer.parseInt(datas.substring(pos, pos + 2), 16);
                double illu = MqttTools.parseHex(datas.substring(pos + 2, pos + 2 + len * 2)) / 10;
                //光合作用强度 photosynthesis
                pos = pos + 2 + len * 2 + 2;
                len = Integer.parseInt(datas.substring(pos, pos + 2), 16);
                double phot = MqttTools.parseHex(datas.substring(pos + 2, pos + 2 + len * 2)) / 10;
                //湿度 humidity
                pos = pos + 2 + len * 2 + 2;
                len = Integer.parseInt(datas.substring(pos, pos + 2), 16);
                double humi = MqttTools.parseHex(datas.substring(pos + 2, pos + 2 + len)) / 10;
                //温度 temperature
                double fieldTemp = MqttTools.parseHex(datas.substring(pos + 2 + len, pos + 2 + len * 2)) / 10;
                System.out.println("termid=1," + "光照度：" + illu + "," + "光合作用强度：" + phot + "," + "湿度：" + humi + "," + "温度：" + fieldTemp);
                Map<String, String> tags = new HashMap<String, String>();
                tags.put("type", "fieldTemp");
                Map<String, Object> fields = new HashMap<String, Object>();
                fields.put("value", fieldTemp);
                // 时间使用毫秒为单位
                influxDBConfig.insert("db2", tags, fields, System.currentTimeMillis(), TimeUnit.MILLISECONDS);
                //清空
                tags.clear();
                fields.clear();
                tags.put("type", "phot");
                fields.put("value", phot);
                influxDBConfig.insert("db2", tags, fields, System.currentTimeMillis(), TimeUnit.MILLISECONDS);
                tags.clear();
                fields.clear();
                tags.put("type", "humi");
                fields.put("value", humi);
                influxDBConfig.insert("db2", tags, fields, System.currentTimeMillis(), TimeUnit.MILLISECONDS);
                tags.clear();
                fields.clear();
                tags.put("type", "illu");
                fields.put("value", illu);
                influxDBConfig.insert("db2", tags, fields, System.currentTimeMillis(), TimeUnit.MILLISECONDS);
                System.out.println("insert ok");
            } else if ("8d00003100000002".equals(termid) && !datas.substring(8, 10).equals("00")) {
                //co2
                int pos = 10;
                int len = Integer.parseInt(datas.substring(pos, pos + 2), 16);
                double co2 = MqttTools.parseHex(datas.substring(pos + 2, pos + 2 + len * 2));
                System.out.println("termid=2," + "Co2浓度：" + co2);
                Map<String, String> tags = new HashMap<String, String>();
                tags.put("type", "co2");
                Map<String, Object> fields = new HashMap<String, Object>();
                fields.put("value", co2);
                influxDBConfig.insert("db2", tags, fields, System.currentTimeMillis(), TimeUnit.MILLISECONDS);
                System.out.println("insert ok");
            } else if ("8d00003100000003".equals(termid) && !datas.substring(8, 10).equals("00")) {
                //浊度
                int pos = 10;
                int len = Integer.parseInt(datas.substring(pos, pos + 2), 16);
                double turb = MqttTools.parseHex(datas.substring(pos + 2, pos + 2 + len * 2)) / 100;
                //PH
                pos = pos + 2 + len * 2 + 2;
                len = Integer.parseInt(datas.substring(pos, pos + 2), 16);
                double ph = MqttTools.parseHex(datas.substring(pos + 2, pos + 2 + len * 2)) / 100;
                //溶解氧
                pos = pos + 2 + len * 2 + 2;
                len = Integer.parseInt(datas.substring(pos, pos + 2), 16);
                double do_ = MqttTools.parseHex(datas.substring(pos + 2, pos + 2 + len * 2)) / 100;
                System.out.println("termid=3," + "浊度：" + turb + "," + "PH：" + ph + "," + "溶解氧：" + do_);
                Map<String, String> tags = new HashMap<String, String>();
                tags.put("type", "turb");
                Map<String, Object> fields = new HashMap<String, Object>();
                fields.put("value", turb);
                influxDBConfig.insert("db2", tags, fields, System.currentTimeMillis(), TimeUnit.MILLISECONDS);
                tags.clear();
                fields.clear();
                tags.put("type", "ph");
                fields.put("value", ph);
                influxDBConfig.insert("db2", tags, fields, System.currentTimeMillis(), TimeUnit.MILLISECONDS);
                tags.clear();
                fields.clear();
                tags.put("type", "do");
                fields.put("value", do_);
                influxDBConfig.insert("db2", tags, fields, System.currentTimeMillis(), TimeUnit.MILLISECONDS);
                System.out.println("insert ok");
            } else if ("8d00003100000004".equals(termid) && !datas.substring(8, 10).equals("00")) {
                //氨氮
                int pos = 10;
                int len = Integer.parseInt(datas.substring(pos, pos + 2), 16);
                double an = MqttTools.parseHex(datas.substring(pos + 2, pos + 2 + len * 2));
                //硝氮
                pos = pos + 2 + len * 2 + 2;
                len = Integer.parseInt(datas.substring(pos, pos + 2), 16);
                double nn = MqttTools.parseHex(datas.substring(pos + 2, pos + 2 + len * 2));
                System.out.println("termid=4," + "氨氮：" + an + "," + "硝氮：" + nn);
                Map<String, String> tags = new HashMap<String, String>();
                tags.put("type", "an");
                Map<String, Object> fields = new HashMap<String, Object>();
                fields.put("value", an);
                // 时间使用毫秒为单位
                influxDBConfig.insert("db2", tags, fields, System.currentTimeMillis(), TimeUnit.MILLISECONDS);
                //清空
                tags.clear();
                fields.clear();
                tags.put("type", "nn");
                fields.put("value", nn);
                influxDBConfig.insert("db2", tags, fields, System.currentTimeMillis(), TimeUnit.MILLISECONDS);
                System.out.println("insert ok");
            }
        }
    }
}
